package com.qch.kafka.demo;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;

/**
 * @Author qch
 * @Date 2021/6/6
 */
public class MyTransactional {
    public static KafkaProducer<String,String>getProducer(){
        Map<String, Object> configs = new HashMap<String, Object>();
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.142.135:9092");
        configs.put("key.serializer",StringSerializer.class);
        configs.put("value.serializer",StringSerializer.class);
        configs.put(ProducerConfig.CLIENT_ID_CONFIG,"tx_producer_01");
        //使用事务就必须设置
        configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "tx_id_02");
        configs.put(ProducerConfig.ACKS_CONFIG, "all");
        //是否启动幂等性
        configs.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(configs);
        return producer;

    }
    public static KafkaConsumer<String,String> getConsumer(String groupid)
    {
        Map<String, Object> configs = new HashMap<String, Object>();
        configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.142.135:9092");
        configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configs.put(ConsumerConfig.GROUP_ID_CONFIG, groupid);
        //不启用消费者偏移量的自动确认,也不使用手动确认
        configs.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        configs.put(ConsumerConfig.CLIENT_ID_CONFIG, "consumer_client_02");
        configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(configs);
        return consumer;
    }

    public static void main(String[] args) {
        String connsumerGroupID="consumer_grp_02";
        KafkaProducer<String, String> producer = getProducer();
        KafkaConsumer<String, String> consumer = getConsumer(connsumerGroupID);
        producer.initTransactions();
        consumer.subscribe(Collections.singleton("tp_tx_01"));
        ConsumerRecords<String, String> records = consumer.poll(1_000);
        producer.beginTransaction();
        try{
        HashMap<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
        for (ConsumerRecord<String, String> record : records) {
            System.out.println(record);
            producer.send(new ProducerRecord<String, String>("tp_tx_out_01",record.key(),record.value()));
            offsets.put(
                    new TopicPartition(record.topic(),record.partition()),
                    new OffsetAndMetadata(record.offset()+1)//偏移量表示下一条要消费的消息
            );
        }
        //将该消息的偏移量提交作为事务的一部分，随事务提交和【回滚（不提交消费偏移量）】
        producer.sendOffsetsToTransaction(offsets,connsumerGroupID);
      //  int i=1/0;
        producer.commitTransaction();
        }catch (Exception e){
            e.printStackTrace();
            producer.abortTransaction();
        }finally {
            producer.close();
            consumer.close();
        }

    }
}
